package com.atguigu.realtime.apps

import java.time.LocalDate
import java.util
import java.util.Properties

import com.alibaba.fastjson.JSON
import com.atguigu.realtime.apps.AlertDemo.appName
import com.atguigu.realtime.beans.{OrderDetail, OrderInfo, ProvinceInfo, SaleDetail, UserInfo}
import com.atguigu.realtime.constants.{DBNameConstant, PrefixConstant, TopicConstant}
import com.atguigu.realtime.utils.{DStreamUtil, DateTimeUtil, JedisUtil, PropertiesUtil}
import com.google.gson.Gson
import org.apache.kafka.clients.consumer.ConsumerRecord
import org.apache.spark.SparkConf
import org.apache.spark.broadcast.Broadcast
import org.apache.spark.sql.{DataFrame, Dataset, SparkSession}
import org.apache.spark.streaming.dstream.{DStream, InputDStream}
import org.apache.spark.streaming.kafka010.{CanCommitOffsets, HasOffsetRanges, OffsetRange}
import org.apache.spark.streaming.{Seconds, StreamingContext}
import redis.clients.jedis.Jedis

import scala.collection.mutable
import scala.collection.mutable.ListBuffer

/**
 * Created by Smexy on 2022/8/26
 *
 * 整体思路:
 * 下单事实(order_detail) + 用户维度 + 省份维度 + 商品维度
 *
 * `order_detail` LEFT JOIN `order_info` : 有user_id,province_id
 * 再使用user_id 关联 user_info
 * 再使用province_id 关联 `base_province`
 *
 * -------------------------------------
 * order_detail 和 order_info 是事实表:  只有下单事实产生时，数据才会产生。
 *      需要使用canal实时采集mysql的binlog,获取新下单的数据，生产到kafka中。
 *
 * user_info： 维度表。 会缓慢变化(update 和 insert)的维度表。
 *      每次关联时，必须关联最新的用户信息(mysql)。
 *      把Mysql中的user_info同步到redis(关联时，关联redis，性能好)!
 *
 *
 * base_province： 维度表。 几乎不变的维度表，更类似编码表。
 *      使用时，可以一次读取，缓存在App中。不需要频繁关联数据库。
 *      广播变量(Driver端定义一个只读变量，广播到Executor)
 *
 *
 * ---------------------------------------------
 *    只有 DS[(K,V)] 才能Join
 *
 *        select  order_detail` LEFT JOIN `order_info`  on order_detail.orderId = order_info.id
 *
 *  -----------------------------
 *    双流Join，有以下几个限制:
 *         ①只有 DS[(K,V)]才能Join
 *         ②两个流，必须从同一个StreamingContext获取，意味 batchDuration一致
 *         ③只有同一个批次的数据，才能Join。
 *            join无法跨批次
 *
 * -------------------------
 *  无法Join的原因在于 你要Join的数据，位于不同的批次，彼此错过！
 *
 *  -------------------------
 *    解决： 对于数据来说，谁早到，谁就在缓存等着。
 *
 *    数量关系:   1笔订单的order_info 对应 N个order_detail
 *
 *    order_info:
 *          ①   和当前批次已经达到的order_detail 关联
 *          ②   无条件写入缓存，以防后续后迟到的order_detail
 *          ③   去缓存查询，看是否有早到的order_detail，有就关联
 *
 *    order_detail:
 *          ④   和当前批次已经达到的order_info 关联(只需要做一次，可以在order_info逻辑中操作)。
 *          ⑤   如果 ④关联不上，说明当前的order_detail早到或晚到了
 *               需要去缓存中，看是否有早到的order_info
 *                  有： 说明order_detail晚到了，就匹配
 *                 ⑥ 无：  说明order_detail早到了。 把自己写入缓存。
 *
 *   ----------------------------
 *    设计redis中 order_info 和order_detail存储的模式。
 *        使用场景：   order_detail 到缓存寻找 order_info时，根据orderId匹配。
 *
 *        order_info:
 *          粒度： 一个Order_info 是一个K-V
 *          k:  orderId
 *          v:  string
 *
 *          读:  get    如果key不存在，返回的value是null
 *          写:  set
 *
 *        使用场景：   order_info 到缓存寻找 order_detail时，根据orderId匹配N个order_detail。
 *        order_detail:
 *          k:  orderId
 *          v: Set
 *
 *          读： smembers  如果key不存在，返回的value是Set(空)
 *          写:  sadd
 */
object SaleDetailApp extends BaseApp {

  override var batchDuration: Int = 10
  override var appName: String = "SaleDetailApp"
  override var groupId: String = "220409realtime"
  override var topic: String = TopicConstant.ORDER_DETAIL
  val topic2: String = TopicConstant.ORDER_INFO

  def queryProvinceInfoFromMysql(sparkConf :SparkConf):mutable.HashMap[String,ProvinceInfo]={

    //批处理   SparkSQL(高阶)  RDD(低阶API)
    val sparkSession: SparkSession = SparkSession.builder().config(sparkConf).getOrCreate()

    val properties = new Properties()
    properties.put("user",PropertiesUtil.getValue("jdbc.user"))
    properties.put("password",PropertiesUtil.getValue("jdbc.password"))

    //查询Mysql
    val dataFrame: DataFrame = sparkSession.read.jdbc(
      PropertiesUtil.getValue("jdbc.url"),
      "base_province",
      properties
    )
    import sparkSession.implicits._

    val ds: Dataset[ProvinceInfo] = dataFrame.as[ProvinceInfo]

    //把查询结果收集到Driver
    val provinceInfoes: util.List[ProvinceInfo] = ds.collectAsList()

    val provinceMap = new mutable.HashMap[String, ProvinceInfo]

    provinceInfoes.forEach(provinceInfo => {
      provinceMap.put(provinceInfo.id,provinceInfo)
    })

    provinceMap

  }

  def main(args: Array[String]): Unit = {

    val sparkConf: SparkConf = new SparkConf().setMaster("local[*]").setAppName(appName)
      .set("es.nodes", PropertiesUtil.getValue("es.nodes"))
      .set("es.port", PropertiesUtil.getValue("es.port"))
      //如果要写入的index不存在，允许自动创建
      .set("es.index.auto.create", "true")
      //允许把hostname转为ip
      .set("es.nodes.wan.only", "true")

    //重写父类中的context
    context = new StreamingContext(sparkConf,Seconds(batchDuration))

    //在Driver端只查询一次。 要注意顺序，一定编写在  new StreamingContext 之下
    val provinceMap: mutable.HashMap[String, ProvinceInfo] = queryProvinceInfoFromMysql(sparkConf)

    //广播出去 广播后一定用广播变量，而不是之前的集合
    val pmBC: Broadcast[mutable.HashMap[String, ProvinceInfo]] = context.sparkContext.broadcast(provinceMap)

    runSparkStreamingApp{

      val orderInfoDS: InputDStream[ConsumerRecord[String, String]] = DStreamUtil.getDStream(context, groupId, topic2)
      val orderDetailDS: InputDStream[ConsumerRecord[String, String]] = DStreamUtil.getDStream(context, groupId, topic)

      var OIoffsetRanges: Array[OffsetRange] = null
      var ODoffsetRanges: Array[OffsetRange] = null
      //获取当前批次的偏移量，将 ConsumerRecord 转换为 Bean
      val ds1: DStream[(String, OrderDetail)] = orderDetailDS.transform(rdd => {

        ODoffsetRanges = rdd.asInstanceOf[HasOffsetRanges].offsetRanges

        rdd.map(record => {
          val orderDetail: OrderDetail = JSON.parseObject(record.value(), classOf[OrderDetail])
          (orderDetail.order_id, orderDetail)
        })

      })

      val ds2: DStream[(String, OrderInfo)] = orderInfoDS.transform(rdd => {

        OIoffsetRanges = rdd.asInstanceOf[HasOffsetRanges].offsetRanges

        rdd.map(record => {

          val orderInfo: OrderInfo = JSON.parseObject(record.value(), classOf[OrderInfo])
          orderInfo.create_date = DateTimeUtil.parseTimeStrToDate(orderInfo.create_time)
          orderInfo.create_hour = DateTimeUtil.parseTimeStrToHour(orderInfo.create_time)

          (orderInfo.id, orderInfo)

        })

      })

      val ds3: DStream[(String, (Option[OrderInfo], Option[OrderDetail]))] = ds2.fullOuterJoin(ds1)

      // ds3是关联的结果
      val ds4: DStream[SaleDetail] = ds3.mapPartitions(partition => {

        val jedis: Jedis = JedisUtil.getJedis

        val gson = new Gson()

        //准备集合，存放封装号的购物详情
        val saleDetails: ListBuffer[SaleDetail] = ListBuffer[SaleDetail]()

        partition.foreach {
          case (orderId, (orderInfoOption, orderDetailOption)) => {

            if (orderInfoOption != None) {

              val orderInfo: OrderInfo = orderInfoOption.get

              //①   和当前批次已经达到的order_detail 关联c
              if (orderDetailOption != None) {

                val orderDetail: OrderDetail = orderDetailOption.get

                saleDetails.append(new SaleDetail(orderInfo, orderDetail))
              }

              //② 无条件写入缓存，以防后续后迟到的order_detail
              jedis.setex(PrefixConstant.order_info_redis_preffix + orderId, PrefixConstant.max_delay_time,gson.toJson(orderInfo))

              //③   去缓存查询，看是否有早到的order_detail，有就关联
              val orderDetails: util.Set[String] = jedis.smembers(PrefixConstant.order_detail_redis_preffix + orderId)

              orderDetails.forEach(jsonStr => {

                val od: OrderDetail = JSON.parseObject(jsonStr, classOf[OrderDetail])

                saleDetails.append(new SaleDetail(orderInfo, od))

              })

            } else {

              val orderDetail: OrderDetail = orderDetailOption.get

              //⑤需要去缓存中，看是否有早到的order_info
              val orderInfoStr: String = jedis.get(PrefixConstant.order_info_redis_preffix + orderId)

              if (orderInfoStr != null) {

                //说明order_info早到了，已经在缓存中等着了
                val oi: OrderInfo = JSON.parseObject(orderInfoStr, classOf[OrderInfo])

                saleDetails.append(new SaleDetail(oi, orderDetail))

              } else {

                //⑥ 无：  说明order_detail早到了。 把自己写入缓存。
                jedis.sadd(PrefixConstant.order_detail_redis_preffix + orderId, gson.toJson(orderDetail))

                jedis.expire(PrefixConstant.order_detail_redis_preffix + orderId ,PrefixConstant.max_delay_time)

              }

            }


          }
        }

        jedis.close()

        saleDetails.toIterator


      })

      //ds4是join成功的数据，但是没有 user和province的维度信息
      val ds5: DStream[SaleDetail] = ds4.mapPartitions(partition => {

        val jedis: Jedis = JedisUtil.getJedis

        val saleDetails: Iterator[SaleDetail] = partition.map(saleDetail => {

          //关联省份
          val pi: ProvinceInfo = pmBC.value.get(saleDetail.province_id).get
          saleDetail.mergeProvinceInfo(pi)

          //关联用户
          val userStr: String = jedis.get(PrefixConstant.user_info_redis_preffix + saleDetail.user_id)

          val userInfo: UserInfo = JSON.parseObject(userStr, classOf[UserInfo])
          saleDetail.mergeUserInfo(userInfo)

          saleDetail


        })

        jedis.close()

        saleDetails

      })

      import org.elasticsearch.spark._
      //写入ES
      ds5.foreachRDD(rdd => {

        rdd.cache()

        println("要写入ES:"+rdd.count())
        /*
          resource: String：  index名
          cfg: scala.collection.Map[String, String]： 加一些配置
                es.mapping.id:  代表要写入的Bean中哪个属性作为当前这条消息的id
         */
        rdd.saveToEs(DBNameConstant.SALEDETAILINDEX+LocalDate.now(),Map("es.mapping.id" -> "order_detail_id"))

        //初始的DS提交偏移量
        orderInfoDS.asInstanceOf[CanCommitOffsets].commitAsync(OIoffsetRanges)
        orderDetailDS.asInstanceOf[CanCommitOffsets].commitAsync(ODoffsetRanges)

      })


    }

  }
}
